package com.bins.reactor.demo1;

import com.google.common.collect.Maps;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.test.StepVerifier;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.Random;
import java.util.function.Consumer;

public class Test {

   public static void main(String[] a) throws InterruptedException {
       // normal();

       // assertion();

       // structure();

       // scheduleInc();

       // mergeFlux();

       // test1();

       // test2();

       // sinkGenerate();

       // randomGenerate();

       // createSink();

       // monoTest1();

       // monoFromCallable();

       // monoFromCompletionStage();

       // monoDelay();

       // monoTest2();

       // monoBuffer();

       // monoWindow();

       // fluxZipWith();

       // fluxTake();

       // fluxMerge();

       // fluxFlatMap();

       // fluxHotData();

       monoCreate();

       // fluxCreate();

   }

    private static void fluxCreate() {
        Flux.just("Hello", "World").subscribe(System.out::println);
        Flux.fromArray(new Integer[] {1, 2, 3}).subscribe(System.out::println);
        Flux.empty().subscribe(System.out::println);
        Flux.range(1, 10).subscribe(System.out::println);
        Flux.interval(Duration.of(10, ChronoUnit.SECONDS)).subscribe(System.out::println);
    }

    private static void monoCreate() {
        Mono mono1 = Mono.just(Maps.newHashMap());

        Mono mono2 = Mono.just(1);

        Mono mono3 = Mono.just("1");

        Mono.justOrEmpty(null);

        Mono.justOrEmpty(1);

        Mono.empty();

        //empty()：创建一个不包含任何元素，只发布结束消息的序列
        Mono.empty().subscribe(System.out::println);
        //just()：可以指定序列中包含的全部元素。创建出来的 Mono序列在发布这些元素之后会自动结束。
        Mono.just("www.xttblog.com").subscribe(System.out::println);
        //ustOrEmpty()：从一个 Optional 对象或可能为 null 的对象中创建 Mono。
        //只有 Optional 对象中包含值或对象不为 null 时，Mono 序列才产生对应的元素。
        Mono.justOrEmpty(null).subscribe(System.out::println);
        Mono.justOrEmpty("业余草").subscribe(System.out::println);
        Mono.justOrEmpty(Optional.of("业余草")).subscribe(System.out::println);
        //error(Throwable error)：创建一个只包含错误消息的序列。
        Mono.error(new RuntimeException("error")).subscribe(System.out::println, System.err::println);
        //never()：创建一个不包含任何消息通知的序列。
        Mono.never().subscribe(System.out::println);
        //通过 create()方法来使用 MonoSink 来创建 Mono。
        Mono.create(sink -> sink.success("业余草")).subscribe(System.out::println);

        //通过fromRunnable创建，并实现异常处理
        Mono.fromRunnable(() -> {
            System.out.println("thread run");
            throw new RuntimeException("thread run error");
        }).subscribe(System.out::println, System.err::println);
        //通过fromCallable创建
        Mono.fromCallable(() -> "callable run ").subscribe(System.out::println);
        //通过fromSupplier创建
        Mono.fromSupplier(() -> "create from supplier").subscribe(System.out::println);

        //delay(Duration duration)和 delayMillis(long duration)：创建一个 Mono 序列，在指定的延迟时间之后，产生数字 0 作为唯一值。
//        long start = System.currentTimeMillis();
//        Disposable disposable = Mono.delay(Duration.ofSeconds(2)).subscribe(n -> {
//            System.out.println("生产数据源："+ n);
//            System.out.println("当前线程ID："+ Thread.currentThread().getId() + ",生产到消费耗时："+ (System.currentTimeMillis() - start));
//        });
//        System.out.println("主线程"+ Thread.currentThread().getId() + "耗时："+ (System.currentTimeMillis() - start));
//        while(!disposable.isDisposed()) { }
    }

    private static void fluxHotData() throws InterruptedException {
        final Flux<Long> source = Flux.interval(Duration.ofSeconds(1))
                .take(10)
                .publish()
                .autoConnect();
        source.subscribe();
        Thread.sleep(5000);
        source
                .toStream()
                .forEach(System.out::println);
    }

    private static void fluxFlatMap() {
        Flux.just(5, 10)
                .flatMap(x -> Flux.interval(Duration.ofSeconds(1)).take(x))
                .toStream()
                .forEach(System.out::println);
    }

    private static void fluxMerge() {
        Flux.merge(Flux.interval(Duration.ofSeconds(1)).take(5), Flux.interval(Duration.ofSeconds(100)).take(5))
                .toStream()
                .forEach(System.out::println);
//        Flux.mergeSequential(Flux.intervalMillis(0, 100).take(5), Flux.intervalMillis(50, 100).take(5))
//                .toStream()
//                .forEach(System.out::println);
    }

    private static void fluxTake() {
        Flux.range(1, 1000).take(10).subscribe(System.out::println);
        Flux.range(1, 1000).takeLast(10).subscribe(System.out::println);
        Flux.range(1, 1000).takeWhile(i -> i < 10).subscribe(System.out::println);
        Flux.range(1, 1000).takeUntil(i -> i == 10).subscribe(System.out::println);
    }

    private static void fluxZipWith() {
        Flux.just("a", "b")
                .zipWith(Flux.just("c", "d"))
                .subscribe(System.out::println);
        Flux.just("a", "b")
                .zipWith(Flux.just("c", "d"), (s1, s2) -> String.format("%s-%s", s1, s2))
                .subscribe(System.out::println);
    }

    private static void monoWindow() {
        Flux.range(1, 100).window(20).subscribe(System.out::println);
        Flux.interval(Duration.ofSeconds(1)).take(2).toStream().forEach(System.out::println);
    }

    private static void monoBuffer() {
        Flux.range(1, 100).buffer(20).subscribe(System.out::println);
        // Flux.interval(100).bufferMillis(1001).take(2).toStream().forEach(System.out::println);
        Flux.range(1, 10).bufferUntil(i -> i % 2 == 0).subscribe(System.out::println);
        Flux.range(1, 10).bufferWhile(i -> i % 2 == 0).subscribe(System.out::println);
    }

    private static void monoTest2() {
        Mono.fromSupplier(() -> "Hello").subscribe(System.out::println);
        Mono.justOrEmpty(Optional.of("Hello")).subscribe(System.out::println);
        Mono.create(sink -> sink.success("Hello")).subscribe(System.out::println);
    }

    private static void monoDelay() {
        Mono.delay(Duration.ofSeconds(3)).doOnNext(System.out::println).block();
    }

    private static void monoFromCompletionStage() {
        // Mono.fromCompletionStage(future).block();
    }

    private static void monoFromCallable() {
        Mono.fromCallable(() -> "9999").subscribe(System.out::println);
    }

    private static void monoTest1() {
        Mono.defer(()->{
            return Mono.error(new RuntimeException());
        }).subscribe();
    }

    private static void createSink() {
        Flux.create(sink -> {
            for (int i = 0; i < 10; i++) {
                sink.next(i);
            }
            sink.complete();
        }).subscribe(System.out::println);
    }

    private static void randomGenerate() {
        final Random random = new Random();
        Flux.generate(ArrayList::new, (list, sink) -> {
            int value = random.nextInt(100);
            list.add(value);
            sink.next(value);
            if (list.size() == 10) {
                sink.complete();
            }
            return list;
        }).subscribe(System.out::println);
    }

    private static void sinkGenerate() {
        Flux.generate(sink -> {
            sink.next("Hello");
            sink.complete();
        }).subscribe(System.out::println);
    }

    private static void test2() {

        Flux.interval(Duration.ofSeconds(2)).doOnNext(System.out::println).blockLast();
    }

    private static void test1() {
        Flux.range(1, 10)
                .timeout(Flux.never(), v -> Flux.never())
                .subscribe(System.out::println);
    }

    private static void mergeFlux() throws InterruptedException {
        LongSubscriber mySubscriber = new LongSubscriber();
        Flux<Long> flux = Flux.interval(Duration.ofSeconds(1)).take(200);
        Flux<Long> flux1 = Flux.interval(Duration.ofSeconds(1)).take(200);
        Flux<Long> flux2 = flux.mergeWith(flux1).doOnRequest(n -> {System.out.println("订阅者发起请求数据，请求长度" + n);});;
        flux2.subscribe(mySubscriber);
        Thread.sleep(100000);

    }

    private static void scheduleInc() throws InterruptedException {
        LongSubscriber mySubscriber = new LongSubscriber();
        Flux<Long> flux = Flux.interval(Duration.ofSeconds(1)).take(200)
                .doOnRequest(n -> {System.out.println("订阅者发起请求数据，请求长度" + n);});
        flux.subscribe(mySubscriber);
        Thread.sleep(10000);
    }

    private static void structure() {

        Flux<String> fluxArr = Flux.fromArray(new String[]{"刘志强","王妍","张三"});

        List<String> list = new ArrayList();
        list.add("刘志强");
        list.add("王妍");
        list.add("张三");
        Flux<String> fluxList = Flux.fromIterable(list);

        Flux<Integer> flux = Flux.range(1, 6);


    }

    private static void assertion() {
        MySubscriber mySubscriber = new MySubscriber();
        Flux<String> flux = Flux.just("刘志强","王妍","张三")
                .doOnRequest(n -> {System.out.println("订阅者发起请求数据，请求长度" + n);});
        flux.subscribe(mySubscriber);
        StepVerifier.create(flux).expectNext("刘志强").expectNext("王妍").expectNext("张三").verifyComplete();
    }

    public static void normal() {
       MySubscriber mySubscriber = new MySubscriber();
       // Flux 实现出版商（Publisher）接口。
       Flux.just("刘志强","王妍","张三")
               .doOnRequest(n -> {System.out.println("订阅者发起请求数据，请求长度" + n);})
               .subscribe(mySubscriber);
   }

}
